package com.example.mq.mqserver.datacenter;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;
import jdk.internal.util.xml.impl.Input;

import java.awt.*;
import java.io.*;
import java.util.LinkedList;
import java.util.Queue;
import java.util.Scanner;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: DongGuoZhen
 * @Date: 2024/02/27/21:01
 * @Description: 针对存储消息的文件进行管理  对硬盘上的消息进行管理
 */
public class MessageFileManager {
    //此处定义一个内部类,表示该队列的统计信息
    static public class Stat {
        public int totalCount; //总消息数量
        public int validCount; //有效的消息数量
    }


    public void init() {
        //这里暂时不需要额外初始化操作,以备后续扩展
    }

    //约定消息文件所在的目录和文件名

    /**
     * 这个方法用来获取到指定队列对应的消息文件所在路径
     *
     * @param queueName 队列名称
     * @return
     */
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    /**
     * 这个方法用来获取该队列的消息数据文件路径
     *
     * @param queueName 队列名称
     * @return
     */
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    /**
     * 这个方法用来获取该队列的消息统计文件路径
     *
     * @param queueName
     * @return
     */
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }


    //针对消息的统计文件进行处理

    /**
     * 读取消息统计文件
     *
     * @param queueName 队列名称
     * @return Stat
     */
    private Stat readStat(String queueName) {
        //由于当前统计文件是文本文件,直接使用scanner读写
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt(); //总消息数量
            stat.validCount = scanner.nextInt(); //有效消息数量
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 向统计文件中写入消息统计信息
     *
     * @param queueName
     * @param stat
     */
    private void writeStat(String queueName, Stat stat) {
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    //创建队列对应的文件和目录
    public void createQueueFiles(String queueName) throws IOException {
        //1.先创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            //不存在,创建
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败! baseDir=" + baseDir.getAbsolutePath());
            }
        }
        //2.创建消息数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }
        //3.创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }
        //4.给消息统计文件设定初始值 0\t0
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    //删除目录和删除文件

    /**
     * 删除队列的目录和文件
     * 队列也是可以被删除的,当队列删除之后,对应的消息文件也要随之删除
     * 先删除里面的文件,再删除目录
     *
     * @param queueName 队列名称
     */
    public void destroyQueueFiles(String queueName) throws IOException {
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if (!ok1 || !ok2 || !ok3) {
            //删除失败 有任意一个删除失败
            throw new IOException("删除队列目录和文件失败! baseDir=" + baseDir.getAbsolutePath());
        }
    }


    //针对队列目录和文件是否存在进行一个检查
    //后续有生产者给brokerServer生成消息了,如果持久化,这个消息会被记录到文件上.
    public boolean checkFilesExits(String queueName) {
        //判定队列的数据文件和统计文件是否都存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            return false;
        }
        return true;
    }

    /**
     * 发送消息
     * 这个方法用来把一个新的消息放在队列对应的文件中
     *
     * @param queue   表示把消息写入队列
     * @param message 要写入的消息
     */
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        //1.检查一下当前要写入队列对应的文件是否存在
        if (!checkFilesExits(queue.getName())) {
            throw new MqException("[MessageFileManager] 队列对应的文件不存在! queueName = " + queue.getName());
        }
        //2.把Message对象进行序列化,转为二进制的字节数组
        byte[] messageBinary = BinaryTool.toBytes(message);
        //加锁,针对队列对象进行加锁
        synchronized (queue) {
            //3.先获取到当前队列数据文件的长度,用这个来计算message对象的offsetBeg和offsetEnd
            //把新的message数据写入到队列数据文件的末尾,此时message对象的offsetBeg就是当前文件长度+4
            //offsetEnd就是当前文件长度+4+message自身长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            //通过这个方法就能获取到文件长度 单位字节  queueDataFile.length();
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            //4.写入消息数据 此处是追加写到数据文件末尾  true表示是追加写
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    //接下来先写当前消息的长度 占据4个字节
                    dataOutputStream.writeInt(messageBinary.length);
                    //写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }
            //5.更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }

    }


    /**
     * 删除消息  逻辑删除  也就是把硬盘上存储的数据里面的isValid 属性设置为0
     * 1.先把文件中这一段数据读出来,还原成message对象
     * 2.把isValid改为0
     * 3.把上述数据重新写入文件
     * 此处这个参数中的message对象,必须包含offsetBeg和offsetEnd
     *
     * @param queue   要删除那个队列中的消息
     * @param message 要删除的消息
     */
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        synchronized (queue) {  //针对queue对象进行加锁
            //使用随机访问文件操作
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                //1,先从文件中读取对应message数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());//光标移动到offsetBeg位置
                randomAccessFile.read(bufferSrc);
                //2. 把当前读到的二进制数据,转换回成Message对象
                //反序列化
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                //3.把isValid设置为无效
                diskMessage.setIsValid((byte) 0x0);
                //4.把diskMessage对象重新写入文件
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                //虽然上面已经seek过了,但是上面seek完了之后,文件进行了读操作,就会导致文件光标向后移动,移动了下一个消息的位置
                //接下来的写入要想写回之前的位置,就需要重新调整文件光标位置
                randomAccessFile.seek(message.getOffsetBeg());//光标移动到这个位置
                randomAccessFile.write(bufferDest);
                //通过上述,对于文件来说,只是有一个字节发生改名而已: isValid= 0x0
            }
            //更新统计文件 一个消息设定为无效了,就需要把消息的有效个数进行-1
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) {
                stat.validCount -= 1;
            }
            writeStat(queue.getName(), stat);
        }
    }


    /**
     * 这个方法在程序启动的时候,进行调用
     * 这个方法从文件中读取出所有的有效消息内容,加载到内存中 放到一个链表中
     * 这里使用linkedList 主要目的是为了后续进行头删操作
     * 这个方法不考虑线程安全问题,因为该方法是在程序启动的时候调用,此时服务器还不能处理请求,不涉及多线程操作
     *
     * @param queueName 队列名称
     * @return 放入在一个链表中
     */
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        //打开文件
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                //使用这个变量记录文件光标
                long currentOffset = 0;
                while (true) {
                    //一个文件中有很多消息,循环读取
                    //1先去读取当前消息长度
                    //这里的readInt方法,读到文件末尾,就抛出一个异常 EOFException
                    int messageSize = dataInputStream.readInt();//先读4个字节  获取当前消息长度  这里的readInt可能会读到文件末尾
                    //2按照这个长度读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);  //读取消息具体长度的内容 返回读取的字节数
                    if (messageSize != actualSize) {
                        //不匹配 说明文件有问题
                        throw new MqException("[MessageFileManager] 文件格式错误! queueName=" + queueName);
                    }
                    //3需要把读到的二进制数据反序列化回Message对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    //4判断这个消息对象是否无效
                    if (message.getIsValid() != 0x1) {
                        //无效数据 直接跳过
                        currentOffset += (4 + messageSize);  //虽然消息无效,但是offset要记得更新
                        continue;
                    }
                    //5有效数据,需要把message对象加入到链表中,加入之前需要填写offsetBeg和offsetEnd
                    //在进行计算offset 的时候,需要知道当前文件光标的位置.
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);
                    messages.add(message);
                }

            } catch (EOFException e) {
                //这里的 catch 是文件读到末尾的时候,会被readInt抛出这个异常
                System.out.println("[MessageFileManager] 将硬盘数据读取到内存中 恢复Message数据完成");
            }
        }
        return messages;
    }


    /**
     * 判断当前是否要针对该队列的消息数据文件进行GC
     * 总数超过2000,并且有效消息低于50%
     *
     * @param queueName 当前队列
     * @return true:进行GC   false:不进行GC
     */
    public boolean checkGC(String queueName) {
        //判断是否要GC 根据总消息数 和 有消息消息数量 在消息统计文件中
        Stat stat = readStat(queueName);
        if (stat.totalCount > 2000 && (double) stat.validCount / (double) stat.totalCount < 0.5) {
            return true;  //进行GC
        }
        return false;
    }

    /**
     * 这个方法就是把新消息先写入这个文件中
     *
     * @param queueName 队列名称
     * @return 要写入新消息的目录
     */
    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "queue_data_new.txt";
    }

    /**
     * 通过这个方法真正执行消息文件的垃圾回收
     * 使用复制算法
     * 创建一个新的文件: queue_data_new.txt
     * 把之前的数据文件中有效的消息读出来,写入到新的文件中
     * 删除旧的文件,再把新文件名改为queue_data.txt
     * 同时更新消息统计文件
     *
     * @param queue 消息队列
     */
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        //进行GC是针对消息数据文件,进行大洗牌,在这个过程中,其他线程不能对该队列的消息文件进行修改
        synchronized (queue) {
            //由于GC操作比较耗时,统计一下执行的消耗时间
            long gcBeg = System.currentTimeMillis();
            //1. 创建一个新的文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNewFile.exists()) {
                //正常情况下,这个文件是不应该存在的,如果存在就是以外,说明上次GC了一半,程序以外崩溃了
                throw new MqException("[MessageFileManager] gc 时发现该对列的 queue_data_new 已经存在! queueName" + queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if (!ok) {
                throw new MqException("[MessageFileManager] 创建文件失败! queueDataNewFile= " + queueDataNewFile.getAbsolutePath());
            }

            //2. 从旧的文件中读取出所有的有效消息(这里直接调用上述方法即可)
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            //3. 把有效消息写入到新的文件中
            try(OutputStream outputStream = new FileOutputStream(queueDataNewFile)) {
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        byte[] buffer = BinaryTool.toBytes(message);
                        //先写4个字节消息的长度
                        dataOutputStream.writeInt(buffer.length);
                        //再写入消息内容
                        dataOutputStream.write(buffer);
                    }
                }
            }
            //4.删除旧的数据文件,并且把新的进行重命名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if(!ok) {
                throw new MqException("[MessageFileManager] 删除旧文件失败! queueDataOldFile = " + queueDataOldFile.getAbsolutePath());
            }
            // 把queue_data_new.txt 重命名为 queue_data.txt
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok) {
                throw new MqException("[MessageFileManager] 文件重命名失败! queueDataNewFile = " + queueDataNewFile.getAbsolutePath()+
                        ",queueDataOlaFile = "+ queueDataOldFile.getAbsolutePath());
            }
            //5.更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(),stat);
            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc执行完毕! queueName" + queue.getName() + ", time= "
                    + (gcEnd - gcBeg) + "ms");
        }
    }
}
